package com.aotain.nyx.allattstatd;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map.Entry;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

import com.aotain.apollo.mongodb.MongoImportTool;
import com.aotain.common.mongo.DataColumn;
import com.aotain.nyx.ipstat.IPStatTuple;
import com.aotain.nyx.statis.AbnStatisTuple;

public class AllAttDStatisImport implements FlatMapFunction<Tuple2<String,AbnStatisTuple>	, Object> {

	/** 
	 * @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	 */ 
	private static final long serialVersionUID = 6706009868030895225L;

	@Override
	public void flatMap(Tuple2<String, AbnStatisTuple> value,
			Collector<Object> out) throws Exception {
		// TODO Auto-generated method stub
		String tableName = "SDS_ALL_ATT_D";
		AbnStatisTuple col = value.f1;
		if(col == null)
			return;
		String dip = col.getDestIP();
		String sip = col.getSourceIP();
		String type = col.getType();
		String lastTime = col.getAttTime();
		String sourcePro = col.getSourceProvince();
		String attType = col.getAttackType();
		String typeStr = "";
		switch (type) {
			case "SESSIONMIDDLE" : 
				typeStr = "流量异常";
				break;
			case "REQUESTABNL" :
				typeStr = "流量异常";
				break;
			case "ACKFLOOD" :
				typeStr = "Ack Flood";
				break;
			case "BOTS" :
				typeStr = "僵尸主机";
				break;
			case "CC" :
				typeStr = "Web攻击";
				break;
			case "COMMANDINJECTION" :
				typeStr = "命令注入";
				break;
			case "DNSATTACK" :
				typeStr = "DNS Flood";
				break;
			case "ICMPFLOOD" :
				typeStr = "ICMP Flood";
				break;
			case "PORTLOW" :
				typeStr = "端口异常";
				break;
			case "SCANATTACK" :
				typeStr = "扫描攻击";
				break;
			case "SENSITIVEURL" :
				typeStr = "敏感路径";
				break;
			case "SQLINJECTION" :
				typeStr = "sql注入";
				break;
			case "SYNFLOOD" :
				typeStr = "Syn Flood";
				break;
			case "UDPFLOOD" :
				typeStr = "UDP Flood";
				break;
			case "VC" :
				typeStr = "暴力破解";
				break;
			case "XSS" :
				typeStr = "XSS 跨站脚本";
				break;
			default :
				typeStr = type;
				break;
		}

		List<DataColumn> saverow = new ArrayList<DataColumn>();
		DataColumn colDIP = new DataColumn();
		colDIP.setKey("DESTIP");
		colDIP.setValue(dip);
		saverow.add(colDIP);
		DataColumn colSIP = new DataColumn();
		colSIP.setKey("SOURCEIP");
		colSIP.setValue(sip);
		saverow.add(colSIP);
		DataColumn colType = new DataColumn();
		colType.setKey("THREATTYPE");
		colType.setValue(typeStr);
		saverow.add(colType);
		DataColumn colLastTime = new DataColumn();
		colLastTime.setKey("LASTTIME");
		colLastTime.setValue(Long.parseLong(lastTime));
		saverow.add(colLastTime);
		DataColumn colSourcePro = new DataColumn();
		colSourcePro.setKey("SOURCEPROVINCE");
		colSourcePro.setValue(sourcePro);
		saverow.add(colSourcePro);
		DataColumn colAttType = new DataColumn();
		colAttType.setKey("ATTTYPE");
		colAttType.setValue(attType);
		saverow.add(colAttType);

		SimpleDateFormat sdfd = new SimpleDateFormat("yyyyMMdd");
		Date date = new Date();
		String dateDay = sdfd.format(date);
		String reporttime = dateDay;

		DataColumn colReportTime = new DataColumn();
		colReportTime.setKey("REPORTTIME");
		colReportTime.setValue(Long.parseLong(reporttime));
		saverow.add(colReportTime);

		List<DataColumn> incrow = new ArrayList<DataColumn>();
		DataColumn colFlow = new DataColumn();
		colFlow.setKey("ATTNUM");
		colFlow.setValue(col.getAttackNum());
		incrow.add(colFlow);
		DataColumn colOctets = new DataColumn();
		colOctets.setKey("ATTOCTETS");
		colOctets.setValue(value.f1.getUpStreamOctets());
		incrow.add(colOctets);
		DataColumn colPacket = new DataColumn();
		colPacket.setKey("ATTPACKET");
		colPacket.setValue(value.f1.getUpStreamPacket());
		incrow.add(colPacket);
		

		List<DataColumn> filter = new ArrayList<DataColumn>();
		filter.add(new DataColumn("REPORTTIME",Long.parseLong(reporttime)));
		filter.add(new DataColumn("DESTIP",dip));
		filter.add(new DataColumn("SOURCEIP",sip));
		filter.add(new DataColumn("THREATTYPE",typeStr));
//		filter.add(new DataColumn("ATTTYPE",attType));


		MongoImportTool.getInstance().UpdateRowData(tableName, filter, saverow);

		MongoImportTool.getInstance().IncRowData(tableName, filter, incrow);
	}

}
